package cn._51doit.live.jobs;

import cn._51doit.live.proto.DataBeanProto;
import cn._51doit.live.utils.FlinkUtilsV2;
import cn._51doit.live.utils.PBSerializer;
import cn._51doit.live.utils.ProtoBufDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStream;

/**
 * 从Kafka中读取protobuf类型的二进制数据，并且将二进制数据转成protobuf的bean类型
*/
public class GiftCountFromProtoBuf {

    public static void main(String[] args) throws Exception {

        //注册DataBeanProto.DataBean使用ProtoBuf进行序列化
        FlinkUtilsV2.env.getConfig().registerTypeWithKryoSerializer(DataBeanProto.DataBean.class, PBSerializer.class);


        DataStream<DataBeanProto.DataBean> beanStream = FlinkUtilsV2.createKafkaStream(args[0], ProtoBufDeserializationSchema.class);

        beanStream.print();

        FlinkUtilsV2.env.execute();

    }
}